package com.wzy.chapter05;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

public class TransformFilterTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 1. 自定义类实现
        SingleOutputStreamOperator<Event> result1 = stream.filter(new MyFilter());
        // 2. 匿名类实现
        SingleOutputStreamOperator<Event> result2 = stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event event) throws Exception {
                return Objects.equals(event.user, "Bob");
            }
        });
        // 3. lambda实现
        SingleOutputStreamOperator<Event> result3 = stream.filter(data -> data.user.equals("Bob"));

        result1.print("1");
        result2.print("2");
        result3.print("3");
        env.execute();
    }

    public static class MyFilter implements FilterFunction<Event> {
        @Override
        public boolean filter(Event event) throws Exception {
            return Objects.equals(event.user, "Bob");
        }
    }
}
